publish() uses a single-space MessageBody sentinel when the payload is empty, but SqsReceivedMessage base64-decodes with validate=True and falls back to UTF-8 bytes on failure. This makes an empty payload round-trip as b" " instead of b"". Consider treating the sentinel body (e.g., exactly " ") as an empty payload when parsing, or use a different encoding strategy that preserves empty bytes while satisfying SQS' non-empty body requirement.

SqsReceivedMessage currently attempts strict base64 decode and on failure returns the UTF-8 bytes of the raw body. Since SqsServer.publish()/reply() use a single space as a sentinel for empty payloads, this code will decode that sentinel into b" " instead of an empty payload. Special-case the sentinel body (exactly " ") to produce b"" before attempting base64 decode.

In the detached reject() task created during cancellation, _ignore_exception only suppresses Exception. If the reject task itself is cancelled, task.result() raises asyncio.CancelledError (a BaseException), which can surface as an unhandled exception in the event loop's done-callback handling. Suppress asyncio.CancelledError here as well (as is already done in _consume_channel).

This test cancels the background task but never awaits it. That can lead to "Task was destroyed but it is pending" warnings (or failures under stricter asyncio settings). After task.cancel(), await the task under contextlib.suppress(asyncio.CancelledError) (or ensure it finishes) before ending the test.

is_active is driven solely by the _active flag, but _active is never reset when the main consume task finishes on its own (e.g., when channels_to_callbacks is empty and _consume() returns immediately). This can leave a subscriber reporting active even though its task is already done, and resume() won’t restart it. Consider defining is_active based on both _active and _main_task state (e.g., _main_task is not None and not _main_task.done()), and/or ensuring _consume() sets _active = False on exit.

When receive_wait_time_seconds is configured as 0 (used in tests and potentially by users), an empty receive response results in a tight loop that repeatedly calls receive_message() with no backoff. This can cause unnecessary CPU usage and request spam. Consider adding a small sleep/yield when messages is empty and wait time is 0 (or enforcing a minimum wait time) to prevent busy looping.

repid/connections/sqs/__init__.py uses a plain from ... import SqsServer plus a string-based __all__. In this codebase, other connection subpackages re-export with from ... import <Name> as <Name> (e.g., repid/connections/redis/__init__.py:1-2, repid/connections/nats/__init__.py:1) which avoids unused-import linting and keeps the pattern consistent. Consider switching to from repid.connections.sqs.message_broker import SqsServer as SqsServer and dropping __all__ to match the established convention.